package org.lzfto.flink.demo.business.student;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ExecutionException;

import org.apache.flink.core.execution.JobClient;
import org.apache.flink.runtime.minicluster.MiniClusterJobClient;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.lzfto.flink.demo.common.result.PubResult;
import org.lzfto.flink.demo.business.flinkTest.base.db.BasicJdbcTable;
import org.lzfto.flink.demo.utils.FlinkUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

@Lazy
@Service
public class StudentService implements StudentRepository {
    private final Logger LOGGER = LoggerFactory.getLogger(getClass());
    
    @Lazy
    @Autowired
    private JdbcTemplate jdbcTemplate;

    private static final String INSERT_SQL = 
        "INSERT INTO student (student_id, name, gender, birth_date, email) VALUES (?, ?, ?, ?, ?)";
    
    private static final String SELECT_BY_ID_SQL = 
        "SELECT student_id, name, gender, birth_date, email, last_optime FROM student WHERE student_id = ?";
    
    private static final String SELECT_ALL_SQL = 
        "SELECT student_id, name, gender, birth_date, email, last_optime FROM student WHERE 1=1";
    
    private static final String UPDATE_SQL = 
        "UPDATE student SET name = ?, gender = ?, birth_date = ?, email = ? WHERE student_id = ?";
    
    private static final String DELETE_SQL = 
        "DELETE FROM student WHERE student_id = ?";

    @Override
    public void create(Student student) {
        try {
            LOGGER.info("Creating student with ID: {}", student.getStudentId());
            int affectedRows = jdbcTemplate.update(INSERT_SQL,
                student.getStudentId(),
                student.getName(),
                student.getGender().name(),
                student.getBirthDate(),
                student.getEmail());
                
            if (affectedRows == 0) {
                LOGGER.error("Failed to create student: {}", student.getStudentId());
                throw new RuntimeException("Failed to create student: " + student.getStudentId());
            }
            LOGGER.info("Successfully created student with ID: {}", student.getStudentId());
        } catch (Exception e) {
            LOGGER.error("Error creating student with ID: {} - {}", student.getStudentId(), e.getMessage());
            throw new RuntimeException("Failed to create student: " + student.getStudentId(), e);
        }
    }

    @Override
    public Student findById(int studentId) {
        try {
            LOGGER.debug("Fetching student with ID: {}", studentId);
            Student student = jdbcTemplate.queryForObject(
                SELECT_BY_ID_SQL,
                new BeanPropertyRowMapper<>(Student.class),
                studentId);
            LOGGER.debug("Successfully fetched student with ID: {}", studentId);
            return student;
        } catch (Exception e) {
            LOGGER.error("Error fetching student with ID: {} - {}", studentId, e.getMessage());
            throw new RuntimeException("Failed to find student by id: " + studentId, e);
        }
    }

    @Override
    public List<Student> findAll(Student condition) {
        try {
            LOGGER.debug("Fetching students with condition: {}", condition);
            
            StringBuilder sqlBuilder = new StringBuilder(SELECT_ALL_SQL);
            List<Object> params = new ArrayList<>();
            
            if (condition != null) {
                if (condition.getStudentId() > 0) {
                    sqlBuilder.append(" AND student_id = ?");
                    params.add(condition.getStudentId());
                }
                if (condition.getName() != null && !condition.getName().isEmpty()) {
                    sqlBuilder.append(" AND name LIKE ?");
                    params.add("%" + condition.getName() + "%");
                }
                if (condition.getGender() != null) {
                    sqlBuilder.append(" AND gender = ?");
                    params.add(condition.getGender().name());
                }
                if (condition.getEmail() != null && !condition.getEmail().isEmpty()) {
                    sqlBuilder.append(" AND email LIKE ?");
                    params.add("%" + condition.getEmail() + "%");
                }
            }
            
            List<Student> students = jdbcTemplate.query(
                sqlBuilder.toString(),
                new BeanPropertyRowMapper<>(Student.class),
                params.toArray());
                
            LOGGER.debug("Successfully fetched {} students", students.size());
            return students;
        } catch (Exception e) {
            LOGGER.error("Error fetching students - {}", e.getMessage());
            throw new RuntimeException("Failed to find students", e);
        }
    }

    @Override
    public void update(Student student) {
        try {
            LOGGER.info("Updating student with ID: {}", student.getStudentId());
            int affectedRows = jdbcTemplate.update(UPDATE_SQL,
                student.getName(),
                student.getGender().name(),
                student.getBirthDate(),
                student.getEmail(),
                student.getStudentId());
            
            if (affectedRows == 0) {
                LOGGER.error("No student found with ID: {}", student.getStudentId());
                throw new RuntimeException("No student found with id: " + student.getStudentId());
            }
            LOGGER.info("Successfully updated student with ID: {}", student.getStudentId());
        } catch (Exception e) {
            LOGGER.error("Error updating student with ID: {} - {}", student.getStudentId(), e.getMessage());
            throw new RuntimeException("Failed to update student: " + student.getStudentId(), e);
        }
    }

    @Override
    public void delete(int studentId) {
        try {
            LOGGER.info("Deleting student with ID: {}", studentId);
            int affectedRows = jdbcTemplate.update(DELETE_SQL, studentId);
            
            if (affectedRows == 0) {
                LOGGER.error("No student found with ID: {}", studentId);
                throw new RuntimeException("No student found with id: " + studentId);
            }
            LOGGER.info("Successfully deleted student with ID: {}", studentId);
        } catch (Exception e) {
            LOGGER.error("Error deleting student with ID: {} - {}", studentId, e.getMessage());
            throw new RuntimeException("Failed to delete student: " + studentId, e);
        }
    }

	public PubResult<List<Map<String, Object>>> sumByGender() {
        // StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env); // 来自于bridge
        // env.setParallelism(1);
        
        //只能使用批处理模式，否则查询返回的记录数很多。
        EnvironmentSettings env = EnvironmentSettings.newInstance().inBatchMode()        		
        		.build();
        TableEnvironment tabEnv = TableEnvironment.create(env);
        // 目标表
        // Flink 中 字段类型需要严格匹配，否则会出现许多奇怪的问题
        // 很可惜内置的函数不能用于create table语法的字段定义部分
        String targetWith = null;
        try {
            targetWith = BasicJdbcTable.getTargetJdbcTable("student").withConnector("jdbc").build();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        String targetTableDDL = """
				create table student_target(
				   student_id BIGINT,
				   name STRING,
				   gender STRING,
				   birth_date DATE,
				   email STRING
				)""" + targetWith;
        tabEnv.executeSql(targetTableDDL);
        Table sumTable = tabEnv.sqlQuery("select count(*) as total, gender from student_target group by gender");
        TableResult result=sumTable.execute();
        try {
            result.await();
            List<Map<String, Object>> data=FlinkUtil.printTable(result);
            Optional<JobClient> client= result.getJobClient();
            if(client.isPresent()){
                //MiniClusterJobClient
                MiniClusterJobClient jClient= (MiniClusterJobClient) client.get();
                String jobId=client.get().getJobID().toString();
                LOGGER.info("flink jobId:{},clusterType:{}",jobId,jClient.getClass().getName());
            }
            return PubResult.ok("success", data);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }
}
